package topic;

import com.aruba.utils.RBConnectionUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;

public class Consumer {

    @Test
    public void consumer1() throws Exception {
        //1. 获取连接对象
        Connection connection = RBConnectionUtil.getConnection();

        //2. 构建信道
        Channel channel = connection.createChannel();

        //3. 构建队列 与生产者相同  参数：队列名 是否持久化 是否排外（只允许一个消费者） 长时间未使用是否自动删除 其他参数
        channel.queueDeclare(Publisher.QUEUE_NAME1, false, false, false, null);

        //4. 监听消息
        DefaultConsumer callback = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1获取到消息：" + new String(body, "UTF-8"));
            }
        };
        // 参数：队列名 是否自动ack 监听回调
        channel.basicConsume(Publisher.QUEUE_NAME1, true, callback);

        System.in.read();
    }

    @Test
    public void consumer2() throws Exception {
        //1. 获取连接对象
        Connection connection = RBConnectionUtil.getConnection();

        //2. 构建信道
        Channel channel = connection.createChannel();

        //3. 构建队列 与生产者相同  参数：队列名 是否持久化 是否排外（只允许一个消费者） 长时间未使用是否自动删除 其他参数
        channel.queueDeclare(Publisher.QUEUE_NAME2, false, false, false, null);

        //4. 监听消息
        DefaultConsumer callback = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2获取到消息：" + new String(body, "UTF-8"));
            }
        };
        // 参数：队列名 是否自动ack 监听回调
        channel.basicConsume(Publisher.QUEUE_NAME2, true, callback);

        System.in.read();
    }

}
